package com.yanggu.flink.datastream_api.window

import cn.hutool.core.date.DateUtil
import com.yanggu.flink.datastream_api.pojo.SensorReading
import com.yanggu.flink.datastream_api.source.MySensorSource
import org.apache.flink.configuration.{Configuration, RestOptions}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date

/**
 * 滚动窗口(Tumbling Window): 根据数据的时间依据固定的窗口长度对数据进行切分, 将数据划分到不同的时间段中, 有窗口的开始时间和结束时间
 * 特点: 时间对齐, 窗口长度固定, 没有重叠
 * 适用场景：适合做 BI 统计等（做每个时间段的聚合计算）
 * 滚动窗口TumblingEventTimeWindows.of(size, offset), 窗口的大小和偏移量
 */
object TumblingWindowDemo {

  def main(args: Array[String]): Unit = {

    //创建本地执行环境, 并且拥有WebUi和设置端口
    val config = new Configuration()
    config.setInteger(RestOptions.PORT.key(), 8080)
    val environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)

    environment
      .addSource(new MySensorSource)
      //设置事件时间升序的Watermark
      .assignAscendingTimestamps(_.timestamp)
      //根据id进行分区
      .keyBy(_.id)
      //创建了事件时间滚动窗口, 窗口大小为1minute
      .window(TumblingEventTimeWindows.of(Time.minutes(1L)))
      //执行全窗口函数, 将窗口内的数据集收集齐后执行process方法
      .process(new ProcessWindowFunction[SensorReading, String, String, TimeWindow] {
        override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[String]): Unit = {
          val result = s"当前时间: ${DateUtil.formatDateTime(new Date)}, " +
            s"滚动窗口开始时间: ${DateUtil.formatDateTime(new Date(context.window.getStart))}, 窗口结束时间: ${DateUtil.formatDateTime(new Date(context.window.getEnd))}, " +
            s"传感器id: $key" + s"过去1minute, 总共有${elements.size}条传感器数据"
          out.collect(result)
        }
      })
      .print("result ")

    environment.execute("TumblingWindowDemo Job")
  }

}
